package com.hanson.demo.queue.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import com.hanson.demo.actors.Message;
import com.hanson.demo.queue.TbQueueConsumer;
import com.hanson.demo.service.UserActorService;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

@Component
@Slf4j
@RequiredArgsConstructor
public class KafkaConsumer implements TbQueueConsumer<Message> {

	private final UserActorService userActorService;
	// kafka的监听器，topic为"hanson"
	@KafkaListener(topics = "hanson", groupId = "laiGroup")
	public void consumerMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
		String value = record.value();
		this.consumerMessage(record.topic(), value);
		// 手动提交offset
		ack.acknowledge();
	}

	// 配置多个消费组
	/**
	 * 我们可以使用 groupId 属性来指定Kafka消费者组。同一消费者组中的消费者会共享Kafka主题中的消息，从而实现分布式消息的消费。
	 */
	@KafkaListener(topicPattern = "lai.aa.*", groupId = "hansonGroup")
	public void listenTulingGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
		String value = record.value();
		log.info("==>listenTulingGroup:{}", value);
		ack.acknowledge();
	}

	@Override
	public void consumerMessage(String topic, String msg) {
		log.info("==>consumerMessage:topic{},msg:{}", topic, msg);
		userActorService.dealQueueConsumer(topic,msg);
	}

	@Override
	public boolean isStopped() {
		return false;
	}

}
